package com.dz.timer;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;

import java.util.List;
import java.util.Random;

/**
 * @ClassName MySourceTuple3
 * @Description TODO
 * @Author zhangdong
 * @Date 2020/11/14 18:36
 * @Version 1.0
 */
public class MySourceTuple3 implements SourceFunction<Tuple3<String,Long,Long>> {

     boolean runing=true;

     Random random =new Random();
    private List<String> names= Lists.newArrayList("张","李","周","田","王","彭");
    long num=1;
    @Override
    public void run(SourceContext<Tuple3<String, Long, Long>> ctx) throws Exception {
        while (runing)
        {
            int next = random.nextInt(6);

          Tuple3<String,Long,Long> tuple3 =  new Tuple3<String,Long,Long>(names.get(next),num,System.currentTimeMillis());
            //ctx.collect(tuple3);
            ctx.collectWithTimestamp(tuple3,tuple3.f2);
            if(tuple3.f2!=null)
            {
                ctx.emitWatermark(new Watermark(
                        tuple3.f2
                ));
            }
            num+=1;
           //  Thread.sleep(10000);
        }
    }

    @Override
    public void cancel() {
          runing=false;
    }
}
